package io.wen.bd.s3m1;

import org.apache.commons.lang.time.StopWatch;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;

/**
 * 解题思路：
 * 使用一个Reduce Task, 保证全局有序
 * Map阶段将文本数字读取转换为IntWritable，且作为key输出，value本身不关注，使用NullWritable
 * Reduce阶段为一个Reduce，接收所有元素，按key排序，定义数字的序号，递增序号作为输出key，数字本身作为value
 *
 * 题目：
 * file1.txt, file2.txt, file3.txt
 * MapReduce 程序读取这三个文件，对三个文件中的数字进行整体升序排序，
 * 并输出到一个结果文件中，结果文件中的每一行有两个数字（两个数字之间使⽤用制表符分隔)，
 * 第一个数字代表排名，第二个数字代表原始数据
 *
 * Usage: %s [genericOptions] <input> <output>
 *
 * 执行结果：resources/oneReduceOut
 * 结果截图：resources/screenshot
 *
 */
public class SortUsingOneReduceTask extends Configured implements Tool {

    @Override
    public int run(String[] args) throws Exception {
        // 校验输入输出目录是否指定
        if (args.length != 2) {
            System.err.printf("Usage: %s [genericOptions] <input> <output> \n\n", this.getClass());
            System.exit(-1);
        }

        // job基础配置
        Job job = Job.getInstance(getConf());
        job.setJobName("SortUsingOneReduceTask");
        job.setJarByClass(getClass());

        // 设置使用1个reduce task，保证全局有序
        job.setNumReduceTasks(1);

        // 设置Mapper、Reducer
        job.setMapperClass(SortMapper.class);
        job.setReducerClass(SortReducer.class);

        // 设置输入输出格式
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(IntWritable.class);

        // 设置输入输出路径
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        return job.waitForCompletion(true) ? 0 : 1;
    }

    /**
     * 排序Mapper
     */
    static class SortMapper extends Mapper<LongWritable, Text, IntWritable, NullWritable> {
        IntWritable num = new IntWritable();
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            // 使用要排序的数字本身作为key, value不需要关注，使用NullWritable
            num.set(Integer.parseInt(value.toString()));
            context.write(num, NullWritable.get());
        }
    }

    /**
     * 排序Reducer
     */
    static class SortReducer extends Reducer<IntWritable, NullWritable, IntWritable, IntWritable> {
        // 定义数字的序号，reduce
        IntWritable index = new IntWritable();
        int curIndex = 1;

        @Override
        protected void reduce(IntWritable key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
            // 定义数字的序号，reduces
            for (NullWritable value : values) {
                index.set(curIndex);
                context.write(index, key);
                curIndex++;
            }
        }
    }

    /**
     * 运行作业，统计时间
     */
    public static void main(String[] args) throws Exception {
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        int exitCode = ToolRunner.run(new SortUsingOneReduceTask(), args);
        stopWatch.stop();
        System.out.println("total use time: " + stopWatch.toString());
        System.exit(exitCode);
    }
}
